package com.atbeijing.D04;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 求5s内每个温感器的最大最小值
 */
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                //.filter(r -> r.id.equals("sensor_1"))
                .keyBy(r -> r.id)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .aggregate(new Agg2(),new ResultWindow2())
                .print();

        env.execute();
    }

    public static class Agg2 implements AggregateFunction<SensorReading, Tuple3<String,Double,Double>,Tuple3<String,Double,Double>>{
        //id,最大值,最小值
        @Override
        public Tuple3<String, Double, Double> createAccumulator() {
            return Tuple3.of("",Double.MIN_VALUE, Double.MAX_VALUE);
        }

        @Override
        public Tuple3<String, Double, Double> add(SensorReading value, Tuple3<String, Double, Double> accumulator) {
            accumulator.f0=value.id;
            accumulator.f1=Double.max(accumulator.f1,value.temperature);
            accumulator.f2=Double.min(accumulator.f1,value.temperature);
            return accumulator;
        }

        @Override
        public Tuple3<String, Double, Double> getResult(Tuple3<String, Double, Double> accumulator) {
            return accumulator;
        }

        @Override
        public Tuple3<String, Double, Double> merge(Tuple3<String, Double, Double> a, Tuple3<String, Double, Double> b) {
            return null;
        }
    }

    public static class ResultWindow2 extends ProcessWindowFunction<Tuple3<String, Double, Double>,Result,String, TimeWindow>{
        @Override
        public void process(String s, Context context, Iterable<Tuple3<String, Double, Double>> iterable, Collector<Result> collector) throws Exception {
            Tuple3<String, Double, Double> next = iterable.iterator().next();
            TimeWindow window = context.window();
            collector.collect(new Result(s,next.f1,next.f2,window.getStart(),window.getEnd()));
        }
    }


    public static class Result{
        public String id;
        public Double max;
        public Double min;
        public Long windowStart;
        public Long windowEnd;

        public Result() {
        }

        @Override
        public String toString() {
            return "Result{" +
                    "id='" + id + '\'' +
                    ", max=" + max +
                    ", min=" + min +
                    ", windowStart=" + windowStart +
                    ", windowEnd=" + windowEnd +
                    '}';
        }

        public Result(String id, Double max, Double min, Long windowStart, Long windowEnd) {
            this.id = id;
            this.max = max;
            this.min = min;
            this.windowStart = windowStart;
            this.windowEnd = windowEnd;
        }
    }

}
